package com.zhihaoliang.coroutine.buss.concurrent

import com.zhihaoliang.coroutine.buss.concurrent.func.AtomicIntegerData
import com.zhihaoliang.coroutine.buss.concurrent.func.CurrentErro
import com.zhihaoliang.coroutine.common.list.ListActivity
import com.zhihaoliang.coroutine.common.list.ListItem
import com.zhihaoliang.coroutine.util.log
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlin.system.measureTimeMillis

/**
 * @author 支豪亮
 * @date 2022/10/31 15:19
 * @des 并行
 */
suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    log("Completed ${n * k} actions in $time ms")
}

class ConcurrentActivity : ListActivity() {


    override val listItem: List<ListItem> = listOf(
        ListItem("The problem", 0),
        ListItem("Volatiles are of no help", 1),
        ListItem("AtomicInteger", 2),
        ListItem("Thread confinement coarse-grained", 3),
        ListItem("Mutual exclusion", 4),
        ListItem("Actor", 5),
    )

    override fun onItemClick(item: ListItem) {
        when (item.index) {
            1 -> CurrentErro.showVolatilesErro(mainScope)
            2 -> AtomicIntegerData.atomicData(mainScope)
            3 -> runSingleThread(mainScope)
            4 -> useMutual(mainScope)
            5 -> useActor(mainScope)
            else -> CurrentErro.showErro(mainScope)
        }
    }

    val counterContext = newSingleThreadContext("CounterContext")

    private fun runSingleThread(mainScope: CoroutineScope) = mainScope.launch(Dispatchers.Default){
        var counter = 0
        withContext(counterContext) {
            massiveRun {
                counter++
            }
        }
        log("Counter = $counter")
    }

    val mutex = Mutex()
    private fun useMutual(mainScope: CoroutineScope) = mainScope.launch(Dispatchers.Default){
        var counter = 0
        massiveRun {
            // protect each increment with lock
            mutex.withLock {
                counter++
            }
        }
        log("Counter = $counter")
    }

    // 计数器 Actor 的各种类型
    sealed class CounterMsg
    object IncCounter : CounterMsg() // 递增计数器的单向消息
    class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() // 携带回复的请求

    // 这个函数启动一个新的计数器 actor
    private fun CoroutineScope.counterActor() = actor<CounterMsg> {
        var counter = 0 // actor 状态
        for (msg in channel) { // 即将到来消息的迭代器
            when (msg) {
                is IncCounter -> counter++
                is GetCounter -> msg.response.complete(counter)
            }
        }
    }

    private fun useActor(mainScope: CoroutineScope) = mainScope.launch(Dispatchers.Default){
        val counter = counterActor() // 创建该 actor
        withContext(Dispatchers.Default) {
            massiveRun {
                counter.send(IncCounter)
            }
        }
        // 发送一条消息以用来从一个 actor 中获取计数值
        val response = CompletableDeferred<Int>()
        counter.send(GetCounter(response))
        log("Counter = ${response.await()}")
        counter.close() // 关闭该actor
    }


}